﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Reflection;
using System.IO;
using Go;
using GoRpc;

namespace GoRpcTest
{
    class Program
    {
        //跨进程传输消息，默认序列化后顺序匹配，如果远端和本地定义不一致将失败
        //跨进程消息必须继承自Rpc.ParamTuple
        //不同进程使用同一份消息定义文件即可，避免依赖同一个数据集才能序列化
        [Rpc.Methods.AllowBind]
        class Msg1 : Rpc.ParamTuple
        {
            [Rpc.Methods.AllowBind(rename: "v1")]
            public int v1;
        }

        //序列化后名称匹配
        [Match("Msg2")]
        class Msg2 : Rpc.ParamTuple
        {
            [Name("v1")]//定义序列化匹配名。同名称的变量类型必须一致
            public int v1;
        }

        [Match("Msg3")]
        class Msg3 : Msg2//可继承跨进程消息
        {
            [Name("v2")]
            public Msg1 v2;//可嵌套定义消息

            [Name("v3")]
            public Msg2[] v3;//可嵌套定义消息数组，可多维，只支持简单数组结构
        }

        [Rpc.Methods.AllowBind]
        class remote_obj
        {
            //绑定可被远端访问的变量，且设置为只能Set，不能Get
            [Rpc.Methods.AllowBind(getter: false)]
            int value1 = 0;

            //绑定可被远端访问的变量，且重命名，如果代码需要混淆，必须重命名
            [Rpc.Methods.AllowBind(rename: "value2")]
            int value2 = 0;

            //绑定可被远端访问的数组，远端可直接读写数组元素
            [Rpc.Methods.AllowBind(rename: "value3")]
            int[][,] value3 = new int[][,] { new int[2, 2], new int[2, 2] };

            //绑定可被远端访问的方法，且重命名，如果代码需要混淆，必须重命名
            [Rpc.Methods.AllowBind(rename: "add")]
            async Task<int> add()
            {
                await generator.sleep(1000);
                return value1 + value2;
            }

            [Rpc.Methods.AllowBind]
            async Task<int> sub()
            {
                await generator.sleep(1000);
                return value1 - value2;
            }

            [Rpc.Methods.AllowBind]
            async Task<int> recv()
            {
                //接收远端发送的字节流数据，这是和远端同步收发的，必须立即进行，否则阻塞套接字，超时后主动关闭
                var recvs = Rpc.Session.OverRecv();
                int msgLen = 0, offset = 0;
                for (int i = 0; i < recvs.Length; i++)
                {
                    msgLen += (int)recvs[i].Length;
                }
                byte[] buffer = new byte[msgLen];
                for (int i = 0; i < recvs.Length; i++)
                {
                    //判断一次完成或者取消后退出接收
                    while (!recvs[i].Completed)
                    {
                        await recvs[i].Recv(buffer, (int)(offset + recvs[i].Offset), (int)(recvs[i].Length - recvs[i].Offset));
                    }
                    offset += (int)recvs[i].Length;
                }
                Console.WriteLine(Encoding.UTF8.GetString(buffer, 0, buffer.Length));
                return recvs.Length;
            }

            [Rpc.Methods.AllowBind]
            Msg1[] msga = null;

            [Rpc.Methods.AllowBind]
            async Task<Msg3> msg(Msg2 msgb)
            {
                await generator.sleep(1000);
                return new Msg3 { v1 = 1, v2 = msga[0], v3 = new Msg2[] { msgb } };
            }
        }

        static void Main(string[] args)
        {
            Rpc.ParamTuple.AssertName();
            Rpc.Methods.AssertName();

            work_service work = new work_service();
            work_strand strand = new work_strand(work);
            generator.go(strand, async delegate ()
            {
                socket_tcp.acceptor accep = new socket_tcp.acceptor();
                if (!accep.bind("127.0.0.1", 2001))
                {
                    return;
                }
                socket_tcp socket = new socket_tcp();
                if (!await accep.accept(socket))
                {
                    return;
                }
                accep.close();
                generator.children sessions = new generator.children();
                Rpc.Session sess = new Rpc.Session(strand, socket, new Rpc.Methods());
                sess.Methods.BindObj("server", new remote_obj());//绑定一个对象，让远端直接访问
                sess.Methods.Bind("done", () => sess.Close());
                sessions.go(async delegate ()
                {
                    await sess.Call("client.set@value1", 123);//设置远端一个变量
                    await sess.Call("client.set@value2", 456);
                    await sess.Call("client.set@value3[0][0,1]", 789);//设置远端数组元素
                    Rpc.Result<int> r1 = await sess.Call<int>("client.add");//调用远端一个方法
                    Rpc.Result<int> r2 = await sess.Call<int>("client.sub");
                    Rpc.Result<int> r3 = await sess.Call<int>("client.get@value3[0][0,1]");//获取远端数组元素
                    Rpc.Result<int[]> r4 = await sess.Call<int[]>("client.len@value3[0]");//获取远端数组元素长度
                    Console.WriteLine($"Server.Add {r1.result}");
                    Console.WriteLine($"Server.Sub {r2.result}");
                    Console.WriteLine($"Server.Arr {r3.result}");
                    Console.WriteLine($"Server.Len {r4.result[0]},{r4.result[1]}");
                });
                sessions.go(async delegate ()
                {
                    //不分段，使用内核模式发送文件给远端。
                    using (FileStream file = File.OpenRead("..\\..\\Program.cs"))
                    {
                        sess.OverSend(file.SafeFileHandle, 0, file.Length);
                        Rpc.Result<int> r3 = await sess.Call<int>("client.recv");
                        Console.WriteLine($"Server.Recv1 {r3.result}");
                    }
                });
                sessions.go(async delegate ()
                {
                    //分两次，每次32字节再分段，使用内核模式发送文件给远端。
                    using (FileStream file = File.OpenRead("..\\..\\App.config"))
                    {
                        long len1 = file.Length / 2;
                        long len2 = file.Length - len1;
                        sess.OverSend(file.SafeFileHandle, 0, len1, 32);
                        sess.OverSend(file.SafeFileHandle, len1, len2, 32);
                        Rpc.Result<int> r3 = await sess.Call<int>("client.recv");
                        Console.WriteLine($"Server.Recv2 {r3.result}");
                    }
                });
                sessions.go(async delegate ()
                {
                    //分两次，每次32字节再分段，回调方式发送文件给远端。
                    using (FileStream file = File.OpenRead("..\\..\\App.config"))
                    {
                        long len1 = file.Length / 2;
                        long len2 = file.Length - len1;
                        Rpc.SendDoubleBuffer dbuff = new Rpc.SendDoubleBuffer(32);//双缓冲模式，读取文件和发送文件同时进行
                        dbuff.Async(() => file.Read(dbuff.ReadBuff, 0, (int)Math.Min(dbuff.ReadBuff.Length, len1)));//首先异步读取一段
                        sess.OverSend(len1, async delegate (long offset, long count)
                        {//第一次使用内存模式发送
                            await dbuff.Exchange();//等待读取完成，并交换读写缓存
                            if (dbuff.Error)
                            {//读取错误后，直接返回将退出本轮发送，远端将收到Cancel消息
                                return;
                            }
                            ArraySegment<byte> data = dbuff.SendData;
                            if (count - data.Count > 0)
                            {//再次异步读取
                                dbuff.Async(() => file.Read(dbuff.ReadBuff, 0, (int)Math.Min(dbuff.ReadBuff.Length, count - data.Count)));
                            }
                            await Rpc.Session.Send(data);//发送一笔数据
                        });
                        sess.OverHandler(async delegate ()
                        {//第一次发送结束回调
                            Console.WriteLine("Server.Send len1 Over");
                            await generator.sleep(1000);
                        });
                        sess.OverSend(len2, delegate (long offset, long count)
                        {//第二次使用内核模式发送
                            return Rpc.Session.Send(file.SafeFileHandle, len1 + offset, (int)Math.Min(32, count));
                        });
                        sess.OverHandler(delegate ()
                        {//第二次发送结束回调
                            Console.WriteLine("Server.Send len2 Over");
                        });
                        Rpc.Result<int> r3 = await sess.Call<int>("client.recv");
                        Console.WriteLine($"Server.Recv3 {r3.result}");
                    }
                });
                sessions.go(async delegate ()
                {
                    //定义一个临时远端回调，让远端调用，返回后失效
                    await sess.Call("Callback", sess.OverCallback(async delegate (int a, int b)
                    {
                        await generator.sleep(1000);
                        return a + b;
                    }));
                });
                sess.Start();
                await sess.Join();
                Console.WriteLine($"Server.Disconnected {sess.Code}");
                await sessions.wait_all();
                sess.Close();
            });
            generator.go(strand, async delegate ()
            {
                await generator.sleep(100);
                socket_tcp socket = new socket_tcp();
                if (await socket.connect("127.0.0.1", 2001))
                {
                    Rpc.Session sess = new Rpc.Session(strand, socket, new Rpc.Methods());
                    sess.Methods.BindObj("client", new remote_obj());
                    sess.Methods.Bind("Callback", async delegate (Rpc.Methods.Name callback)
                    {
                        //调用远端的回调
                        Rpc.Result<int> r1 = await Rpc.Session.Self.Call<int>(callback, 123, 456);
                        Console.WriteLine($"Client.Callback {r1.result}");
                    });
                    generator.children sessions = new generator.children();
                    sessions.go(async delegate ()
                    {
                        await sess.Call("server.set@value1", 456);
                        await sess.Call("server.set@value2", 123);
                        Rpc.Result<int> r1 = await sess.Call<int>("server.add");
                        Rpc.Result<int> r2 = await sess.Call<int>("server.sub");
                        Console.WriteLine($"Client.Add {r1.result}");
                        Console.WriteLine($"Client.Sub {r2.result}");

                        //自定义消息收发
                        await sess.Call("server.set@msga", new Msg1[] { new Msg1 { v1 = 0 } });
                        await sess.Call("server.msga[0].set@v1", 2);
                        Rpc.Result<Msg3> r3 = await sess.Call<Msg3>("server.msg", new Msg2 { v1 = 3 });
                        Console.WriteLine($"Client.Msg {r3.result.v1},{r3.result.v2.v1},{r3.result.v3[0].v1}");
                        await sess.Call("done");
                    });
                    sess.Start();
                    await sess.Join();
                    Console.WriteLine($"Client.Disconnected {sess.Code}");
                    await sessions.stop();
                    sess.Close();
                }
            });
            work.run();
            Console.WriteLine("work done");
            Console.ReadKey();
        }
    }
}
